﻿///////////////////////////////////////////////////////////////////////////////
//
//  This source is part of the  "Managed Task Framework.NET Runtime"  (MTF.NET Runtime).
//
//
//  Copyright (c) 2011, Multicore Consulting Limited (UK)
//  All rights reserved.
//  http://www.multicoreconsulting.co.uk
//
//  Redistribution and use in source and binary forms, with or without
//  modification, are permitted provided that the following conditions
//  are met:
//
//  -  Redistributions of source code must retain the above copyright
//     notice, this list of conditions and the following disclaimer.
//  -  Redistributions in binary form must reproduce the above copyright
//     notice, this list of conditions and the following disclaimer in
//     the documentation and/or other materials provided with the
//     distribution.
//  -  Neither the name of Multicore Consulting Limited nor the names of
//     its contributors may be used to endorse or promote products
//     derived from this software without specific prior written
//     permission.
//
//  THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
//  "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
//  LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS
//  FOR A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE
//  COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT,
//  INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING,
//  BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES;
//  LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER
//  CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT
//  LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN
//  ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
//  POSSIBILITY OF SUCH DAMAGE.
//
///////////////////////////////////////////////////////////////////////////////

using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Diagnostics;
using MTFOrg.MTF.MTFRuntime;
using MTFOrg.MTF.MTFCore.Messages;

namespace MTFOrg.MTF.MTFCore
{
    /// <summary>
    /// Describes modes for advancing the active slot.
    /// </summary>
    [MTFRuntime.IsPropertyEnumeration()]
    public enum SlotAdvanceMode
    {
        AdvanceSingle,
        AdvanceAll
    }

    /// <summary>
    /// Class implementing the producer-consumer object.
    /// </summary>
    partial class ProducerConsumer
    {
        /// <summary>
        /// Defines the mapping between elements and links for a specific connection.
        /// </summary>
        public interface IConnectionMapping
        {
            /// <summary>
            /// Initializes any state that is required for the mapping process.
            /// </summary>
            /// <param name="connection">Connection to be mapped</param>
            /// <param name="consumerDims">Dimensions of the consumer</param>
            /// <param name="producerDims">Dimensions of the producer</param>
            /// <returns>Mapping state object</returns>
            void Initialize(IConnection connection, int[] consumerDims, int[] producerDims);

            /// <summary>
            /// Finds the producer link number connected to the given consumer link number
            /// for the specified connection by using the mapping provided by the connection.
            /// </summary>
            /// <param name="consumerElemNum">Element number at the consumer end</param>
            /// <param name="consumerLinkNum">Consumer link number within the given connection</param>
            /// <param name="producerElemNum">Element number at the producer end</param>
            /// <param name="producerLinkNum">Producer link number within the given connection</param>
            void MapConsumerToProducer(int consumerElemNum, int consumerLinkNum, out int producerElemNum, out int producerLinkNum);

            /// <summary>
            /// Retrieves connection dimensions from the perspective of a consumer element.
            /// </summary>
            /// <returns>Connection dimensions</returns>
            int[] GetConsumerConnectionDimensions();

            /// <summary>
            /// Retrieves connection dimensions from the perspective of a producer element.
            /// </summary>
            /// <returns>Connection dimensions</returns>
            int[] GetProducerConnectionDimensions();

            /// <summary>
            /// Gets whether this connection has any disabled links.
            /// </summary>
            /// <returns>true if some links may be disabled, otherwise false</returns>
            bool HasDisabledLinks();

            /// <summary>
            /// Gets whether the specified consumer link on the specified element is disabled.
            /// </summary>
            /// <param name="consumerElemNum">Consumer element number</param>
            /// <param name="consumerLinkNum">Consumer link number</param>
            /// <returns>true if link is disabled, otherwise false</returns>
            bool IsConsumerLinkDisabled(int consumerElemNum, int consumerLinkNum);

            /// <summary>
            /// Gets whether the specified producer link on the specified element is disabled.
            /// </summary>
            /// <param name="producerElemNum">Producer element number</param>
            /// <param name="producerLinkNum">Producer link number</param>
            /// <returns>true if link is disabled, otherwise false</returns>
            bool IsProducerLinkDisabled(int producerElemNum, int producerLinkNum);
        }

        public abstract class ConnectionMapping : IConnectionMapping
        {
            /// <summary>
            /// Initializes any state that is required for the mapping process.
            /// </summary>
            /// <param name="connection">Connection to be mapped</param>
            /// <param name="consumerDims">Dimensions of the consumer</param>
            /// <param name="producerDims">Dimensions of the producer</param>
            /// <returns>Mapping state object</returns>
            public abstract void Initialize(IConnection connection, int[] consumerDims, int[] producerDims);

            /// <summary>
            /// Finds the producer link number connected to the given consumer link number
            /// for the specified connection by using the mapping provided by the connection.
            /// </summary>
            /// <param name="consumerElemNum">Element number at the consumer end</param>
            /// <param name="consumerLinkNum">Consumer link number within the given connection</param>
            /// <param name="producerElemNum">Element number at the producer end</param>
            /// <param name="producerLinkNum">Producer link number within the given connection</param>
            public abstract void MapConsumerToProducer(int consumerElemNum, int consumerLinkNum, out int producerElemNum, out int producerLinkNum);

            /// <summary>
            /// Retrieves connection dimensions from the perspective of a consumer element.
            /// </summary>
            /// <returns>Connection dimensions</returns>
            public abstract int[] GetConsumerConnectionDimensions();

            /// <summary>
            /// Retrieves connection dimensions from the perspective of a producer element.
            /// </summary>
            /// <returns>Connection dimensions</returns>
            public abstract int[] GetProducerConnectionDimensions();

            /// <summary>
            /// Gets whether this connection has any disabled links.
            /// </summary>
            /// <returns>true if some links may be disabled, otherwise false</returns>
            public virtual bool HasDisabledLinks()
            {
                return false;
            }

            /// <summary>
            /// Gets whether the specified consumer link on the specified element is disabled.
            /// </summary>
            /// <param name="consumerElemNum">Consumer element number</param>
            /// <param name="consumerLinkNum">Consumer link number</param>
            /// <returns>true if link is disabled, otherwise false</returns>
            public virtual bool IsConsumerLinkDisabled(int consumerElemNum, int consumerLinkNum)
            {
                return false;
            }

            /// <summary>
            /// Gets whether the specified producer link on the specified element is disabled.
            /// </summary>
            /// <param name="producerElemNum">Producer element number</param>
            /// <param name="producerLinkNum">Producer link number</param>
            /// <returns>true if link is disabled, otherwise false</returns>
            public virtual bool IsProducerLinkDisabled(int producerElemNum, int producerLinkNum)
            {
                return false;
            }
        }

        /// <summary>
        /// Defines a simple mapping between elements and links for a specific connection.
        /// This is a one-to-one mapping where the Nth element of the consumer is mapped
        /// to the Nth element of the producer.
        /// </summary>
        public class SimpleConnectionMapping : ConnectionMapping
        {
            /// <summary>
            /// Available link mapping modes.
            /// </summary>
            enum MappingType
            {
                FlatLinear,
                FlatCross,
                Multidimensional
            }

            /// <summary>
            /// Class for holding mapping state.
            /// </summary>
            class MappingInfo
            {
                public MappingType mappingType;

                public int[] consumerDims;
                public string[] consumerDimNames;

                public int[] producerDims;
                public string[] producerDimNames;

                public int[] consumerCxnDims;
                public string[] consumerCxnDimNames;
                public int[] producerCxnDims;
                public string[] producerCxnDimNames;
            }

            /// <summary>
            /// Mapping information
            /// </summary>
            MappingInfo m_MapInfo;

            /// <summary>
            /// Calculates the connection names and sizes of the connection from the consumer
            /// perspective, given names and sizes of the producer and consumer dimensions.
            /// </summary>
            private void CalculateConsumerCxnDims()
            {
                List<int> cxnDimsList = new List<int>();
                List<string> cxnDimNamesList = new List<string>();

                // Include every dimension of the producer that does not exist in the consumer
                for (int i = 0; i < m_MapInfo.producerDimNames.Length; i++)
                {
                    bool foundInConsumer = false;
                    for (int j = 0; j < m_MapInfo.consumerDimNames.Length; j++)
                    {
                        if (m_MapInfo.producerDimNames[i] == m_MapInfo.consumerDimNames[j])
                        {
                            foundInConsumer = true;
                            break;
                        }
                    }
                    if (!foundInConsumer)
                    {
                        cxnDimsList.Add(m_MapInfo.producerDims[i]);
                        cxnDimNamesList.Add(m_MapInfo.producerDimNames[i]);
                    }
                }

                if (cxnDimsList.Count == 0)
                {
                    cxnDimsList.Add(1);
                    cxnDimNamesList.Add(null);
                }

                m_MapInfo.consumerCxnDims = cxnDimsList.ToArray();
                m_MapInfo.consumerCxnDimNames = cxnDimNamesList.ToArray();
            }

            /// <summary>
            /// Calculates the connection names and sizes of the connection from the producer
            /// perspective, given names and sizes of the producer and consumer dimensions.
            /// </summary>
            private void CalculateProducerCxnDims()
            {
                List<int> cxnDimsList = new List<int>();
                List<string> cxnDimNamesList = new List<string>();

                // Include every dimension of the consumer that does not exist in the producer
                for (int i = 0; i < m_MapInfo.consumerDimNames.Length; i++)
                {
                    bool foundInProducer = false;
                    for (int j = 0; j < m_MapInfo.producerDimNames.Length; j++)
                    {
                        if (m_MapInfo.consumerDimNames[i] == m_MapInfo.producerDimNames[j])
                        {
                            foundInProducer = true;
                            break;
                        }
                    }
                    if (!foundInProducer)
                    {
                        cxnDimsList.Add(m_MapInfo.consumerDims[i]);
                        cxnDimNamesList.Add(m_MapInfo.consumerDimNames[i]);
                    }
                }

                if (cxnDimsList.Count == 0)
                {
                    cxnDimsList.Add(1);
                    cxnDimNamesList.Add(null);
                }

                m_MapInfo.producerCxnDims = cxnDimsList.ToArray();
                m_MapInfo.producerCxnDimNames = cxnDimNamesList.ToArray();
            }

            /// <summary>
            /// Initializes any state that is required for the mapping process.
            /// </summary>
            /// <param name="connection">Connection to be mapped</param>
            /// <param name="consumerDims">Dimensions of the consumer</param>
            /// <param name="producerDims">Dimensions of the producer</param>
            public override void Initialize(IConnection connection, int[] consumerDims, int[] producerDims)
            {
                m_MapInfo = new MappingInfo();

                // Try to retrieve dimension mapping string
                ConnectionProperty dimensionMappingProp = connection.GetProperty("DimensionMapping") as ConnectionProperty;
                string dimensionMapping = string.Empty;
                if (dimensionMappingProp != null)
                    dimensionMapping = dimensionMappingProp.Value as string;

                if (dimensionMapping == string.Empty)
                {
                    // Assume 1-dimensional mapping
                    int numProducerElems = DimensionHelper.GetNumElementsFromDimensions(producerDims);
                    int numConsumerElems = DimensionHelper.GetNumElementsFromDimensions(consumerDims);

                    if (numProducerElems == numConsumerElems)
                    {
                        // N-to-N linear mapping
                        m_MapInfo.mappingType = MappingType.FlatLinear;
                        m_MapInfo.consumerCxnDims = m_MapInfo.producerCxnDims = new int[] { 1 };
                    }
                    else
                    {
                        // N-to-M cross mapping
                        m_MapInfo.mappingType = MappingType.FlatCross;
                        m_MapInfo.consumerCxnDims = new int[] { numProducerElems };
                        m_MapInfo.producerCxnDims = new int[] { numConsumerElems };
                    }
                }
                else
                {
                    // Multi-dimensional mapping
                    m_MapInfo.mappingType = MappingType.Multidimensional;
                    string[] dimMapSides = dimensionMapping.Split(':');
                    m_MapInfo.consumerDimNames = dimMapSides[0].Replace(" ", string.Empty).Split(',');
                    m_MapInfo.producerDimNames = dimMapSides[1].Replace(" ", string.Empty).Split(',');

                    m_MapInfo.consumerDims = consumerDims;
                    m_MapInfo.producerDims = producerDims;

                    CalculateConsumerCxnDims();
                    CalculateProducerCxnDims();
                }
            }

            /// <summary>
            /// Finds the producer link number connected to the given consumer link number
            /// for the specified connection by using the mapping provided by the connection.
            /// </summary>
            /// <param name="consumerElemNum">Element number at the consumer end</param>
            /// <param name="consumerLinkNum">Consumer link number within the given connection</param>
            /// <param name="producerElemNum">Element number at the producer end</param>
            /// <param name="producerLinkNum">Producer link number within the given connection</param>
            public override void MapConsumerToProducer(int consumerElemNum, int consumerLinkNum, out int producerElemNum, out int producerLinkNum)
            {
                switch (m_MapInfo.mappingType)
                {
                    case MappingType.FlatLinear:
                        {
                            producerElemNum = consumerElemNum;
                            producerLinkNum = consumerLinkNum;
                        }
                        break;
                    case MappingType.FlatCross:
                        {
                            producerElemNum = consumerLinkNum;
                            producerLinkNum = consumerElemNum;
                        }
                        break;
                    case MappingType.Multidimensional:
                        {
                            // Calculate multi-dimensional elements
                            int[] consumerElems = DimensionHelper.GetElementsFromElement(m_MapInfo.consumerDims, consumerElemNum);
                            int[] consumerCxnElems = DimensionHelper.GetElementsFromElement(m_MapInfo.consumerCxnDims, consumerLinkNum);

                            // Map producer elements
                            int[] producerElems = new int[m_MapInfo.producerDims.Length];
                            int[] producerCxnElems = new int[m_MapInfo.producerCxnDims.Length];
                            for (int i = 0; i < m_MapInfo.producerDimNames.Length; i++)
                            {
                                // Try to match directly with consumer elems
                                bool matched = false;
                                for (int j = 0; j < m_MapInfo.consumerDimNames.Length; j++)
                                {
                                    if (m_MapInfo.producerDimNames[i] == m_MapInfo.consumerDimNames[j])
                                    {
                                        producerElems[i] = consumerElems[j];
                                        matched = true;
                                        break;
                                    }
                                }
                                if (matched) continue;

                                // Try to match with consumer connection elems
                                for (int j = 0; j < m_MapInfo.consumerCxnDimNames.Length; j++)
                                {
                                    if (m_MapInfo.producerDimNames[i] == m_MapInfo.consumerCxnDimNames[j])
                                    {
                                        producerElems[i] = consumerCxnElems[j];
                                    }
                                }
                            }

                            // Map producer connection elements
                            for (int i = 0; i < m_MapInfo.producerCxnDimNames.Length; i++)
                            {
                                // Try to match directly with consumer connection elems
                                for (int j = 0; j < m_MapInfo.consumerDimNames.Length; j++)
                                {
                                    if (m_MapInfo.producerCxnDimNames[i] == m_MapInfo.consumerDimNames[j])
                                    {
                                        producerCxnElems[i] = consumerElems[j];
                                        break;
                                    }
                                }
                            }

                            // Flatten back down to single element number and link number
                            producerElemNum = DimensionHelper.GetElementFromElements(m_MapInfo.producerDims, producerElems);
                            producerLinkNum = DimensionHelper.GetElementFromElements(m_MapInfo.producerCxnDims, producerCxnElems);
                        }
                        break;
                    default:
                        {
                            throw new MultipleLinksNotAllowedException("ProducerConsumer.cs: Attempt to use an unsupported mapping type.");
                        }
                }
            }

            /// <summary>
            /// Retrieves connection dimensions from the perspective of a consumer element.
            /// </summary>
            /// <returns>Connection dimensions</returns>
            public override int[] GetConsumerConnectionDimensions()
            {
                return m_MapInfo.consumerCxnDims;
            }

            /// <summary>
            /// Retrieves connection dimensions from the perspective of a producer element.
            /// </summary>
            /// <returns>Connection dimensions</returns>
            public override int[] GetProducerConnectionDimensions()
            {
                return m_MapInfo.producerCxnDims;
            }
        }



        // ---------------------------- DATA MEMBERS --------------------------------

        /// <summary>
        /// Producer consumer state for each element.
        /// </summary>
        protected ProducerConsumerElement[] m_Elements;

        /// <summary>
        /// The number of identical parallel instances of this object to create
        /// given by the product of its dimensions.
        /// </summary>
        protected int m_NumElements;

        /// <summary>
        /// Object dimensions.
        /// </summary>
        protected int[] m_TotalDimensions;

        /// <summary>
        /// The total number of slots available to each connection link.
        /// </summary>
        private int m_NumSlots;

        /// <summary>
        /// Mode by which consumer slots are advanced.
        /// </summary>
        protected SlotAdvanceMode m_ConsumerSlotAdvanceMode;

        /// <summary>
        /// Mode by which producer slots are advanced.
        /// </summary>
        protected SlotAdvanceMode m_ProducerSlotAdvanceMode;

        /// <summary>
        /// Individual element locks.
        /// </summary>
        protected MTFRuntime.SpinLock[] m_TaskElementLocks;

        /// <summary>
        /// Connections to the consumer end.
        /// </summary>
        protected IConnection[] m_ConsumerConnections;

        /// <summary>
        /// Connections to the producer end.
        /// </summary>
        protected IConnection[] m_ProducerConnections;

        /// <summary>
        /// Number of consumer links.
        /// </summary>
        protected int m_NumConsumerLinks;

        /// <summary>
        /// Number of producer links.
        /// </summary>
        protected int m_NumProducerLinks;

        /// <summary>
        /// Gets the number of consumer links.
        /// </summary>
        public int NumConsumerLinks
        {
            get { return m_NumConsumerLinks; }
        }

        /// <summary>
        /// Gets the number of producer links.
        /// </summary>
        public int NumProducerLinks
        {
            get { return m_NumProducerLinks; }
        }


        // ------------------------- TASK INITIALIZATION ----------------------------

        /// <summary>
        /// Called to initialize this task.
        /// </summary>
        public override void OnInitialize()
        {
            base.OnInitialize();

            // Calculate total number of consumer and producer links
            m_NumConsumerLinks = 0;
            foreach (IConnection consumerConnection in ConsumerConnectionPoint.Connections)
                m_NumConsumerLinks += ConnectionHelper.GetConsumerConnectionSize(consumerConnection);

            m_NumProducerLinks = 0;
            int numProducerSlots = 0;
            foreach (IConnection producerConnection in ProducerConnectionPoint.Connections)
            {
                m_NumProducerLinks += ConnectionHelper.GetProducerConnectionSize(producerConnection);
                numProducerSlots += (int)producerConnection.TargetConnectionPoint.ParentTask.GetProperty("OverlayDepth")
                    * ConnectionHelper.GetProducerConnectionSize(producerConnection);
            }

            // Get properties from task's properties store
            m_ConsumerSlotAdvanceMode = (SlotAdvanceMode)GetProperty("ConsumerAdvanceMode");
            m_ProducerSlotAdvanceMode = (SlotAdvanceMode)GetProperty("ProducerAdvanceMode");
            m_NumSlots = (m_ConsumerSlotAdvanceMode == SlotAdvanceMode.AdvanceSingle) ?
                (int)GetProperty("OverlayDepth") : (int)GetProperty("OverlayDepth") * NumConsumerLinks;

            // Calculate total dimensionality
            m_TotalDimensions = DimensionHelper.GetTotalDimsFromTask(ParentTask);

            // Calculate number of elements
            m_NumElements = DimensionHelper.GetNumElementsFromDimensions(m_TotalDimensions);

            // Create elements
            m_Elements = new ProducerConsumerElement[m_NumElements];
            for (int i = 0; i < m_NumElements; i++)
                m_Elements[i] = new ProducerConsumerElement(m_NumSlots, numProducerSlots, m_NumConsumerLinks, m_NumProducerLinks, m_ProducerSlotAdvanceMode);

            // Create task element locks
            m_TaskElementLocks = new MTFRuntime.SpinLock[NumElements];
            for (int i = 0; i < NumElements; i++)
                m_TaskElementLocks[i] = new MTFRuntime.SpinLock();

            // Store connections for each link
            m_ConsumerConnections = new IConnection[m_NumConsumerLinks];
            for (int i = 0; i < m_NumConsumerLinks; i++)
                m_ConsumerConnections[i] = GetConsumerConnectionFromLinkNumber(i);
            m_ProducerConnections = new IConnection[m_NumProducerLinks];
            for (int i = 0; i < m_NumProducerLinks; i++)
                m_ProducerConnections[i] = GetProducerConnectionFromLinkNumber(i);
        }

        /// <summary>
        /// Gets the number of elements.
        /// </summary>
        public int NumElements
        {
            get { return m_NumElements; }
        }

        /// <summary>
        /// Locks a specific element of this task.
        /// </summary>
        public void LockTaskElement(int elementNumber)
        {
            m_TaskElementLocks[elementNumber].Lock();
        }

        /// <summary>
        /// Unlocks a specific element of this task.
        /// </summary>
        public void UnlockTaskElement(int elementNumber)
        {
            m_TaskElementLocks[elementNumber].Unlock();
        }


        // --------------------------- MESSAGE HANDLING -----------------------------

        /// <summary>
        /// Called when an event occurs on a connection to this object.
        /// </summary>
        /// <param name="cfeArgs">Connection-fired event arguments</param>
        protected sealed override void OnMessage(ConnectionFiredEventArgs cfeArgs)
        {
            IMessage msg = cfeArgs.Message;

            if (msg is EventMsg)
            {
                EventMsg eventMsg = msg as EventMsg;

                int elementNumber = eventMsg.ElementNumber;

                LockTaskElement(elementNumber);
                HandleEventMsg(eventMsg, cfeArgs, elementNumber);
                UnlockTaskElement(elementNumber);
            }
            else if (msg is QueryMsg)
            {
                QueryMsg queryMsg = msg as QueryMsg;

                int elementNumber = queryMsg.ElementNumber;

                LockTaskElement(elementNumber);
                HandleQueryMsg(queryMsg, cfeArgs, elementNumber);
                UnlockTaskElement(elementNumber);
            }
            else if (msg is ReleaseMsg)
            {
                ReleaseMsg releaseMsg = msg as ReleaseMsg;

                int elementNumber = releaseMsg.ElementNumber;

                LockTaskElement(elementNumber);
                HandleReleaseMsg(releaseMsg, cfeArgs, elementNumber);
                UnlockTaskElement(elementNumber);
            }
            else if (msg is ReleasedMsg)
            {
                ReleasedMsg releasedMsg = msg as ReleasedMsg;

                int elementNumber = releasedMsg.ElementNumber;

                LockTaskElement(elementNumber);
                HandleReleasedMsg(releasedMsg, cfeArgs, elementNumber);
                UnlockTaskElement(elementNumber);
            }

            base.OnMessage(cfeArgs);
        }

        /// <summary>
        /// Handles arrival of an event message.
        /// </summary>
        /// <param name="msg">Event message</param>
        /// <param name="cfeArgs">Connection event arguments</param>
        protected void HandleEventMsg(EventMsg eventMsg, ConnectionFiredEventArgs cfeArgs, int elementNumber)
        {
            // Convert relative link number to absolute link number
            int absLinkNumber = ConnectionHelper.ConvertRelativeToAbsoluteConsumerLinkNum(cfeArgs.Connection, eventMsg.LinkNumber);

            // Get consumer link and slot sets from event message
            ProducerConsumerElement element = m_Elements[eventMsg.ElementNumber];
            Slots slots = element.GetSlots();
            int consumerSlot = slots.GetConsumerSlotsInUse(absLinkNumber);
            int producerSlot = slots.GetConsumerOppositeSlot(consumerSlot);

            // Link this consumer back to the producer in the adjacent object
            if (consumerSlot != Slots.NO_SLOT)
            {
#if DEBUG
                if (eventMsg.ProducerKey == Slots.NO_SLOT)
                    throw new KeyNotFoundException();
#endif
                slots.SetConsumerExternalSlot(consumerSlot, absLinkNumber, eventMsg.ProducerKey);
            }

            // Handle event arrival
            element.EventReceived(absLinkNumber, m_ConsumerSlotAdvanceMode);
            if (EventReceived != null)
            {
                object elementState = element.Tag;
                EventReceived(absLinkNumber, slots, consumerSlot, producerSlot, eventMsg.Event, ref elementState, elementNumber);
                element.Tag = elementState;
            }
        }

        /// <summary>
        /// Clears the specified link for the specified element number.
        /// </summary>
        /// <param name="consumerLinkNum">Link number of link to clear</param>
        /// <param name="elementNumber">Element number of link to clear</param>
        public void ClearLink(int consumerLinkNum, int elementNumber)
        {
            m_Elements[elementNumber].ClearLink(consumerLinkNum);
        }

        /// <summary>
        /// Handles arrival of a query message.
        /// </summary>
        /// <param name="msg">Query message</param>
        /// <param name="cfeArgs">Connection event arguments</param>
        protected void HandleQueryMsg(QueryMsg queryMsg, ConnectionFiredEventArgs cfeArgs, int elementNumber)
        {
            // Convert relative link number to absolute link number
            int absLinkNumber = ConnectionHelper.ConvertRelativeToAbsoluteProducerLinkNum(cfeArgs.Connection, queryMsg.ProducerLinkNumber);

            // Get producer link and slot sets from query message
            ProducerConsumerElement element = m_Elements[queryMsg.ElementNumber];
            Slots slots = element.GetSlots();
            int producerSlot = element.QueryReceived(absLinkNumber, m_ProducerSlotAdvanceMode);
            int consumerSlot = slots.GetProducerOppositeSlot(producerSlot);

            // Handle query event
            if (QueryReceived != null)
            {
                object elementState = element.Tag;
                QueryReceived(absLinkNumber, slots, consumerSlot, producerSlot, ref elementState, elementNumber);
                element.Tag = elementState;
            }
        }

        /// <summary>
        /// Handles arrival of a release message.
        /// </summary>
        /// <param name="msg">Release message</param>
        /// <param name="cfeArgs">Connection event arguments</param>
        protected void HandleReleaseMsg(ReleaseMsg releaseMsg, ConnectionFiredEventArgs cfeArgs, int elementNumber)
        {
            // Convert relative link number to absolute link number
            int absLinkNumber = ConnectionHelper.ConvertRelativeToAbsoluteProducerLinkNum(cfeArgs.Connection, releaseMsg.ProducerLinkNumber);

            // Get producer link and slot sets from query message
            ProducerConsumerElement element = m_Elements[releaseMsg.ElementNumber];

            Slots slots = element.GetSlots();
            int producerSlot = releaseMsg.ProducerKey;
            int consumerSlot = slots.GetProducerOppositeSlot(producerSlot);

            bool canDelete = element.PreReleaseReceived(absLinkNumber, producerSlot, m_ProducerSlotAdvanceMode);

            // Handle release event
            if (ReleaseReceived != null)
            {
                object elementState = element.Tag;
                ReleaseReceived(
                    absLinkNumber,
                    slots,
                    consumerSlot,
                    producerSlot,
                    releaseMsg.EventToRelease,
                    releaseMsg.ReleaseType == ReleaseMsg.ReleaseTypes.Rollback,
                    canDelete,
                    ref elementState,
                    elementNumber);
                element.Tag = elementState;
            }

            element.PostReleaseReceived(absLinkNumber, releaseMsg.ProducerKey, m_ProducerSlotAdvanceMode, canDelete);

            if (releaseMsg.NeedResponse)
            {
                // Send released notification
                SendReleased(absLinkNumber, producerSlot, elementNumber);
            }
        }
        
        /// <summary>
        /// Handles arrival of a release message.
        /// </summary>
        /// <param name="msg">Release message</param>
        /// <param name="cfeArgs">Connection event arguments</param>
        protected void HandleReleasedMsg(ReleasedMsg releasedMsg, ConnectionFiredEventArgs cfeArgs, int elementNumber)
        {
            // Convert relative link number to absolute link number
            int absLinkNumber = ConnectionHelper.ConvertRelativeToAbsoluteConsumerLinkNum(cfeArgs.Connection, releasedMsg.DstLink);

            // Get consumer link and slot sets from event message
            ProducerConsumerElement element = m_Elements[releasedMsg.ElementNumber];

            object elementState = element.Tag;
            if (ReleasedReceived != null) ReleasedReceived(absLinkNumber, ref elementState, elementNumber);
            element.Tag = elementState;
        }

        public void LinkConsumerAndProducerSlots(Slots slots, int consumerSlot, int producerSlot)
        {
            slots.SetConsumerOppositeSlot(consumerSlot, producerSlot);
            slots.SetProducerOppositeSlot(producerSlot, consumerSlot);
        }


        // ------------------------- CONSUMER INTERFACE -----------------------------

        /// <summary>
        /// Called when a produced event has been received.
        /// </summary>
        /// <param name="consumerLinkNum">Consumer link (absolute) on which the event arrived</param>
        /// <param name="consumerSlots">Consumer slot state information</param>
        /// <param name="producerSlots">Producer slot state information</param>
        /// <param name="eventReceived">Event that was received</param>
        /// <param name="elementState">State associated with the current element</param>
        /// <param name="elementNumber">Number of the current element</param>
        public delegate void OnEvent(int consumerLinkNum, Slots slots, int consumerSlot, int producerSlot, Event receivedEvent, ref object elementState, int elementNumber);
        public OnEvent EventReceived;

        /// <summary>
        /// Sends a query to the connected producer to request a new event for all elements.
        /// </summary>
        /// <param name="consumerLinkNum">Consumer link (absolute) on which to send the query</param>
        /// <param name="producerSlots">Producer slot state information</param>
        /// <returns>true if query was sent, otherwise false</returns>
        public bool SendQueryOnAllElements(int consumerLinkNum, int producerSlot)
        {
            bool failed = false;
            for (int i = 0; i < m_NumElements; i++)
            {
                failed |= !SendQuery(consumerLinkNum, producerSlot, i);
            }
            return !failed;
        }

        /// <summary>
        /// Sends a query to the connected producer to request a new event.
        /// </summary>
        /// <param name="consumerLinkNum">Consumer link (absolute) on which to send the query</param>
        /// <param name="producerSlots">Producer slot state information</param>
        /// <returns>true if query was sent, otherwise false</returns>
        public bool SendQuery(int consumerLinkNum, int producerSlot, int elementNumber)
        {
            int consumerSlot = m_Elements[elementNumber].QuerySent(consumerLinkNum, producerSlot, m_ConsumerSlotAdvanceMode);

            if (consumerSlot != Slots.NO_SLOT)
            {
                Slots slots = m_Elements[elementNumber].GetSlots();

                // Find producer element and link numbers
                IConnection consumerConnection = m_ConsumerConnections[consumerLinkNum];
                int relConsumerLinkNum = consumerLinkNum - ConnectionHelper.GetStartLinkNumberForConsumerConnection(consumerConnection, ConsumerConnectionPoint);
                int relProducerLinkNum, producerElementNumber;
                ConnectionHelper.MapConsumerToProducer(consumerConnection, elementNumber, relConsumerLinkNum, out producerElementNumber, out relProducerLinkNum);

                // Send query
                QueryMsg queryMsg = new QueryMsg(producerElementNumber, relProducerLinkNum);
                TaskManager.QueueMessage(queryMsg, consumerConnection);

                return true;
            }

            return false;
        }

        /// <summary>
        /// Sends a release message to all the connected producers to dispose
        /// of the consumed event(s).
        /// </summary>
        /// <param name="consumerSlot">Consumer slot state information</param>
        /// <param name="eventToRelease">The event that was produced and now requires releasing</param>
        /// <param name="isRollback">Specifies whether this release should rollback rather than commit</param>
        /// <param name="consumerLinkNum">Consumer link (absolute) on which to send the release</param>
        /// <param name="elementNumber">Element number</param>
        /// <param name="needResponse">Set to true if a released response is required</param>
        public void SendRelease(int consumerSlot, Event eventToRelease, bool isRollback, int consumerLinkNum, int elementNumber, bool needResponse)
        {
            Slots slots = m_Elements[elementNumber].GetSlots();

            // Find producer element and link numbers
            IConnection consumerConnection = m_ConsumerConnections[consumerLinkNum];
            int consumerStartLink = ConnectionHelper.GetStartLinkNumberForConsumerConnection(consumerConnection, ConsumerConnectionPoint);
            int relConsumerLinkNum = consumerLinkNum - consumerStartLink;
            int relProducerLinkNum, producerElementNumber;
            ConnectionHelper.MapConsumerToProducer(consumerConnection, elementNumber, relConsumerLinkNum, out producerElementNumber, out relProducerLinkNum);

            // Send release
            int producerSlot = slots.GetConsumerExternalSlot(consumerSlot, consumerLinkNum);
#if DEBUG
            if (producerSlot == Slots.NO_SLOT)
                throw new KeyNotFoundException();
#endif
            ReleaseMsg releaseMsg = new ReleaseMsg(
                eventToRelease,
                isRollback ? ReleaseMsg.ReleaseTypes.Rollback : ReleaseMsg.ReleaseTypes.Commit,
                producerElementNumber, relProducerLinkNum, producerSlot, needResponse);
            TaskManager.QueueMessage(releaseMsg, consumerConnection);

            m_Elements[elementNumber].EventReleased(consumerSlot, consumerLinkNum, m_ConsumerSlotAdvanceMode);
        }

        /// <summary>
        /// Retrieves the consumer connection associated with the specified link number.
        /// </summary>
        /// <param name="consumerLinkNumber">Consumer link number to retrieve connection for</param>
        /// <returns>Consumer connection associated with specified consumer link number</returns>
        public IConnection GetConsumerConnectionFromLinkNumber(int consumerLinkNumber)
        {
            // Iterate over consumer connections, accumulating the link count
            // and when it exceeds the consumerLinkNumber, return the previous cxn.
            int curLinkNum = 0;
            foreach (IConnection consumerConnection in ConsumerConnectionPoint.Connections)
            {
                curLinkNum += ConnectionHelper.GetConsumerConnectionSize(consumerConnection);

                if (curLinkNum > consumerLinkNumber)
                    return consumerConnection;
            }

            return null;
        }


        // ------------------------- PRODUCER INTERFACE -----------------------------

        /// <summary>
        /// Called when a query has arrived.
        /// </summary>
        /// <param name="producerLinkNum">Producer link (absolute) on which the query arrived</param>
        /// <param name="consumerSlots">Consumer slot state information</param>
        /// <param name="producerSlots">Producer slot state information</param>
        /// <param name="elementState">State associated with the current element</param>
        /// <param name="elementNumber">Number of the current element</param>
        public delegate void OnQuery(int producerLinkNum, Slots slots, int consumerSlot, int producerSlot, ref object elementState, int elementNumber);
        public OnQuery QueryReceived;

        /// <summary>
        /// Called when the consumer notifies it has finished with an event.
        /// </summary>
        /// <param name="producerLinkNum">Link that query was received on</param>
        /// <param name="consumerSlots">Consumer slot states (not used)</param>
        /// <param name="producerSlots">Slot states</param>
        /// <param name="eventToRelease">The event to be released</param>
        /// <param name="isRollback">Whether the release was a roll-back</param>
        /// <param name="elementState">State associated with the current element</param>
        /// <param name="elementNumber">Number of the current element</param>
        /// <returns>true if the release marks completion of a slot set</returns>
        public delegate bool OnRelease(int producerLinkNum, Slots slots, int consumerSlot, int producerSlot, Event eventToRelease, bool isRollback, bool deletedSlot, ref object elementState, int elementNumber);
        public OnRelease ReleaseReceived;

        /// <summary>
        /// Called when the producer notifies it has finished with an event.
        /// </summary>
        /// <param name="producerLinkNum">Link that query was received on</param>
        /// <param name="elementState">State associated with the current element</param>
        /// <param name="elementNumber">Number of the current element</param>
        public delegate void OnReleased(int producerLinkNum, ref object elementState, int elementNumber);
        public OnReleased ReleasedReceived;

        /// <summary>
        /// Produces an event to be consumed by the connected consumer.
        /// </summary>
        /// <param name="producerLinkNum">Producer link (absolute) on which to send the event</param>
        /// <param name="newEvent">Event to send</param>
        public void SendEvent(int producerLinkNum, int producerSlot, Event newEvent, int elementNumber)
        {
            // Find consumer element and link numbers
            IConnection producerConnection = m_ProducerConnections[producerLinkNum];
            int relProducerLinkNum = producerLinkNum - ConnectionHelper.GetStartLinkNumberForProducerConnection(producerConnection, ProducerConnectionPoint);
            int relConsumerLinkNum, consumerElementNumber;
            ConnectionHelper.MapProducerToConsumer(producerConnection, elementNumber, relProducerLinkNum, out consumerElementNumber, out relConsumerLinkNum);

            // Send event
            m_Elements[elementNumber].EventSent(producerLinkNum, producerSlot, m_ProducerSlotAdvanceMode);
            EventMsg eventMsg = new EventMsg(consumerElementNumber, relConsumerLinkNum, newEvent, producerSlot);
            TaskManager.QueueMessage(eventMsg, producerConnection);
        }

        /// <summary>
        /// Produces a released notification to be consumed by the connected consumer.
        /// </summary>
        /// <param name="producerLinkNum">Producer link (absolute) on which to send the event</param>
        /// <param name="newEvent">Event to send</param>
        public void SendReleased(int producerLinkNum, int producerSlot, int elementNumber)
        {
            // Find consumer element and link numbers
            IConnection producerConnection = m_ProducerConnections[producerLinkNum];
            int relProducerLinkNum = producerLinkNum - ConnectionHelper.GetStartLinkNumberForProducerConnection(producerConnection, ProducerConnectionPoint);
            int relConsumerLinkNum, consumerElementNumber;
            ConnectionHelper.MapProducerToConsumer(producerConnection, elementNumber, relProducerLinkNum, out consumerElementNumber, out relConsumerLinkNum);

            // Send released notification
            ReleasedMsg releasedMsg = new ReleasedMsg(consumerElementNumber, relConsumerLinkNum, producerSlot);
            TaskManager.QueueMessage(releasedMsg, producerConnection);
        }

        /// <summary>
        /// Retrieves the provider connection associated with the specified link number.
        /// </summary>
        /// <param name="producerLinkNumber">Producer link number to retrieve connection for</param>
        /// <returns>Producer connection associated with specified producer link number</returns>
        public IConnection GetProducerConnectionFromLinkNumber(int producerLinkNumber)
        {
            // Iterate over producer connections, accumulating the link count
            // and when it exceeds the producerLinkNumber, return the previous cxn.
            int curLinkNum = 0;
            foreach (IConnection producerConnection in ProducerConnectionPoint.Connections)
            {
                curLinkNum += ConnectionHelper.GetProducerConnectionSize(producerConnection);

                if (curLinkNum > producerLinkNumber)
                    return producerConnection;
            }

            return null;
        }

        public object GetElementState(int elementNumber)
        {
            return m_Elements[elementNumber].Tag;
        }

        public void SetElementState(int elementNumber, object elementState)
        {
            m_Elements[elementNumber].Tag = elementState;
        }
    }
}